﻿/****************************************************************************************************
	Copyright (C) 2010 RapidWebDev Organization (http://rapidwebdev.org)
	Author: Eunge, Legal Name: Jian Liu, Email: eunge.liu@RapidWebDev.org

	This program is free software: you can redistribute it and/or modify
	it under the terms of the GNU General Public License as published by
	the Free Software Foundation, either version 2 of the License, or
	(at your option) any later version.

	This program is distributed in the hope that it will be useful,
	but WITHOUT ANY WARRANTY; without even the implied warranty of
	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
	GNU General Public License for more details.

	You should have received a copy of the GNU General Public License
	along with this program.  If not, see <http://www.gnu.org/licenses/>.
 ****************************************************************************************************/

using System;
using System.Collections.Generic;
using System.Configuration;
using System.Globalization;
using System.IO;
using System.Linq;
using System.Runtime.Serialization;
using System.Xml;
using System.Xml.Serialization;
using RapidWebDev.Common;
using RapidWebDev.Common.Data;
using RapidWebDev.SearchIndex.Contract;

namespace RapidWebDev.SearchIndex.MSSQLQueue
{
	/// <summary>
	/// The interface to pull indexing messages from somewhere for index service.
	/// </summary>
	public class IndexMessageCoordinator : IIndexMessageCoordinator
	{
		private static readonly DataContractSerializer dataContractSerializer = new DataContractSerializer(typeof(IndexDocMessage));
                
		/// <summary>
		/// Indicates how many index docs should be pulled at a time, defaults to 25.
		/// </summary>
		public int BlockSize { get; set; }

		/// <summary>
		/// Connection string names of databases which are used to queue for indexing by the service.
		/// </summary>
		public IEnumerable<string> ConnectionStringNames { get; private set; }

		/// <summary>
		/// Constructor
		/// </summary>
		/// <param name="connectionStringNames">Connection string names of databases which are used to queue for indexing by the service.</param>
		public IndexMessageCoordinator(IEnumerable<string> connectionStringNames)
		{
			Kit.NotNull(connectionStringNames, "connectionStringNames");
			this.ConnectionStringNames = connectionStringNames;
			this.BlockSize = 25;
		}

		/// <summary>
		/// Get index queues.
		/// </summary>
		/// <returns></returns>
		public IEnumerable<IndexQueue> GetIndexQueues()
		{
			if (this.ConnectionStringNames.Count() == 0)
				return new List<IndexQueue>();

			List<IndexQueue> indexQueue = new List<IndexQueue>();
			foreach (string connectionStringName in this.ConnectionStringNames)
			{
				using (SearchIndexDataContext ctx = CreateDataContext(connectionStringName))
				{
					List<Guid> applicationIds = (from doc in ctx.IndexDocuments
												 where doc.Status < IndexDocumentStatus.Exception
												 select doc.ApplicationId).Distinct().ToList();
					applicationIds.ForEach(applicationId => indexQueue.Add(new IndexQueue(connectionStringName, applicationId)));
				}
			}

			return indexQueue;
		}

		/// <summary>
		/// Read indexing documents from index queue.
		/// </summary>
		/// <param name="indexQueue">Index queue</param>
		/// <returns></returns>
		public IEnumerable<IndexDocMessage> Read(IndexQueue indexQueue)
		{
			Kit.NotNull(indexQueue, "indexQueue");

			using (SearchIndexDataContext ctx = CreateDataContext(indexQueue.IndexQueueId))
			{
				List<IndexDocument> indexDocs = (from doc in ctx.IndexDocuments
												 where doc.ApplicationId == indexQueue.ApplicationId && doc.Status != IndexDocumentStatus.Exception
												 orderby doc.CreatedOn ascending
												 select doc).Take(this.BlockSize).ToList();

				List<IndexDocMessage> indexDocMessages = new List<IndexDocMessage>();
				foreach (IndexDocument indexDoc in indexDocs)
				{
					using (StringReader stringReader = new StringReader(indexDoc.Body))
					using (XmlReader xmlReader = XmlReader.Create(stringReader))
					{
						IndexDocMessage indexDocMessage = dataContractSerializer.ReadObject(xmlReader) as IndexDocMessage;
						indexDocMessage.Id = indexDoc.Id;
						indexDocMessages.Add(indexDocMessage);
					}
				}

				return indexDocMessages;
			}
		}

		/// <summary>
		/// Start indexing on the document.
		/// </summary>
		/// <param name="indexQueue">Index queue</param>
		/// <param name="indexDocMessage">Indexing document</param>
		public void StartIndexing(IndexQueue indexQueue, IndexDocMessage indexDocMessage)
		{
			using (SearchIndexDataContext ctx = CreateDataContext(indexQueue.IndexQueueId))
			{
				IndexDocument indexDoc = (from doc in ctx.IndexDocuments
										  where doc.ApplicationId == indexQueue.ApplicationId && doc.Id == indexDocMessage.Id
										  select doc).FirstOrDefault();

				if (indexDoc == null) return;

				indexDoc.Status = IndexDocumentStatus.Processing;
				indexDoc.IndexingOn = DateTime.UtcNow;
				ctx.SubmitChanges();
			}
		}

		/// <summary>
		/// Complete the indexing for document by id.
		/// </summary>
		/// <param name="indexQueue">Index queue.</param>
		/// <param name="indexDocId">The id of index doc in queue.</param>
		public void Complete(IndexQueue indexQueue, Guid indexDocId)
		{
			if (indexDocId == Guid.Empty) return;

			using (SearchIndexDataContext ctx = CreateDataContext(indexQueue.IndexQueueId))
			{
				IndexDocument indexDoc = (from doc in ctx.IndexDocuments
										  where doc.ApplicationId == indexQueue.ApplicationId
												&& doc.Id == indexDocId
										  select doc).FirstOrDefault();

				if (indexDoc != null)
				{
					ctx.IndexDocuments.DeleteOnSubmit(indexDoc);
					ctx.SubmitChanges();
				}
			}
		}

		/// <summary>
		/// Log an exception for the document by id.
		/// </summary>
		/// <param name="indexQueue">Index queue.</param>
		/// <param name="indexDocId">The id of index doc in queue.</param>
		/// <param name="exception">The exception message.</param>
		public void LogException(IndexQueue indexQueue, Guid indexDocId, string exception)
		{
			if (indexDocId == Guid.Empty) return;

			using (SearchIndexDataContext ctx = CreateDataContext(indexQueue.IndexQueueId))
			{
				IndexDocument indexDoc = (from doc in ctx.IndexDocuments
										  where doc.ApplicationId == indexQueue.ApplicationId
												&& doc.Id == indexDocId
										  select doc).FirstOrDefault();

				if (indexDoc != null)
				{
					indexDoc.Status = IndexDocumentStatus.Exception;
					indexDoc.ExceptionMessage = exception;
					indexDoc.ExceptionOn = DateTime.UtcNow;
					ctx.SubmitChanges();
				}
			}
		}

		private static SearchIndexDataContext CreateDataContext(string connectionStringName)
		{
			Kit.NotNull(connectionStringName, "connectionStringName");
			ConnectionStringSettings connectionStringSettings = ConfigurationManager.ConnectionStrings[connectionStringName];
			if (connectionStringSettings == null)
				throw new ConfigurationErrorsException(string.Format(CultureInfo.InvariantCulture, @"The connection string ""{0}"" is not configured.", connectionStringName));

			return new SearchIndexDataContext(connectionStringSettings.ConnectionString);
		}
	}
}